package com.yanggu.spark.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

//spark-streaming的wordCount程序
object SparkStreaming01_StreamWordCount {

  def main(args: Array[String]): Unit = {

    //1. 初始化Spark配置信息
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming01_StreamWordCount")

    //2. 初始化SparkStreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    //3. 通过监控端口创建DStream, 读进来的数据为一行行
    val lineStreams = ssc.socketTextStream("localhost", 9999)

    //4. 将每一行数据进行切分, 形成一个个单词
    val wordStreams = lineStreams.flatMap(_.split(" "))

    //5. 将单词映射成元祖 (word, 1)
    val wordAndOneStream = wordStreams.map((_, 1))

    //6. 将相同的单词次数进行相加
    val wordCountStreams = wordAndOneStream.reduceByKey(_ + _)

    //7. 打印
    wordCountStreams.print

    //启动采集器
    ssc.start

    //等待采集器的结束
    ssc.awaitTermination


  }

}
